/*
**  Copyright (c) 2014 Michael Liu(michael.liu.point@gmail.com).
**
**  Project: Gateway Unified Management Platform
**  File:    cmsvc.c
**  Author:  Michael
**  Date:    03/20/2014
**
**  Purpose:
**    common management service implement.
*/

/* Include files. */

#include "cmsvc.h"
#include "dpmgr.h"
#include "eamgr.h"
#include "samgr.h"
#include "osdep.h"
#include "pfdep.h"

/* Macro constant definitions. */

#define SEND_MSG_TIMEOUT_SECOND		(5)
#define SEND_MSG_TIMEOUT_USECOND	(0)

#define RECV_MSG_TIMEOUT_SECOND		(5)
#define RECV_MSG_TIMEOUT_USECOND	(0)

#define SESSION_IDLE_CHECK_TIME     (300)

#define SESSION_ID_MAX				(32)
#define SESSION_ID_TMP				(0)

#undef CMSVC_USE_TASK

/* Type definitions. */

typedef struct {
	struct list_head list;
	tcm_req_body *req;
	tcm_rsp_body *rsp;
} tcm_proc_list_node;

/* Local function declarations. */

static e_ret cmsvc_create_session(tcm_session ** session, int svc_socket);
static void cmsvc_destroy_session(tcm_session * session);

static void *cmsvc_task(void *arg);
static e_ret cmsvc_svc_socket_init(int *cmsvc_socket);
static e_ret cmsvc_svc_socket_clean(int cmsvc_socket);
static e_ret cmsvc_proc_session(tcm_session * session);
static e_ret cmsvc_recv_msg(tcm_session * session, struct timeval *timeout);
static e_ret cmsvc_send_msg(tcm_session * session, struct timeval *timeout);
static e_ret cmsvc_proc_msg(tcm_session * session);
static e_ret cmsvc_proc_request(tcm_session * session);

/* Macro API definitions. */

/* Global variable declarations. */

static unsigned char id_table[SESSION_ID_MAX + 1];

e_ret cmsvc_start(void)
{

#ifdef CMSVC_USE_TASK
	int rc;
	pthread_t cmsvc_thread;

	rc = pthread_create(&cmsvc_thread, NULL, cmsvc_task, NULL);
	if (rc != 0) {
		sysmgr_err("pthread_create( cmsvc task ) failed, err: %d.\n", rc);
		return (eRET_FAILURE);
	}
#else /* CMSVC_USE_TASK */
	cmsvc_task(NULL);
#endif /* CMSVC_USE_TASK */

	return (eRET_SUCCESS);

}

void *cmsvc_task(void *arg)
{

	e_ret ret;
	int rc;
	int nfds;
	int cmsvc_socket;
	int select_counter;
	fd_set recv_set;
	tcm_session *session;
	unsigned char rx_buff[4];
	struct timespec ts;
	struct timeval timeout;
	struct list_head session_list;
	struct list_head *each, *next;

	sysmgr_trace_enter();
	sysmgr_dbg("start cmsvc_task()\n");

	INIT_LIST_HEAD(&session_list);

	/* Init socket for cm api. */
	ret = cmsvc_svc_socket_init(&cmsvc_socket);
	if (ret != eRET_SUCCESS) {
		sysmgr_err("cmsvc_svc_socket_init() failed.\n");
		goto out;
	}

	/* select socket, proccess when got a message. */
	while (1) {

		FD_ZERO(&recv_set);
		FD_SET(cmsvc_socket, &recv_set);
		nfds = cmsvc_socket;

		list_for_each(each, &session_list) {
			session = list_entry(each, tcm_session, sibling);
			FD_SET(session->conn_fd, &recv_set);
			if (session->conn_fd > nfds) {
				nfds = session->conn_fd;
			}
		}

		timeout.tv_sec = 1;
		timeout.tv_usec = 0;
		select_counter = select(nfds + 1, &recv_set, NULL, NULL, &timeout);
		if (select_counter == 0) {
			/* Time out... */
			//sysmgr_dbg("select() timeout err: %d,[%s]\n", errno, strerror(errno));
			errno = 0;
			ret = eRET_TIMEOUT;
			continue;
		} else if (select_counter < 0) {
			/* select be interrupt? */
			if (errno != EINTR) {
				//sysmgr_dbg("select() return %d, err: %d,[%s]\n", select_counter,
				//      errno, strerror(errno));
			}
			continue;
		}
		//sysmgr_dbg("select() somethings[%d] are ready.\n", select_counter);

		if (FD_ISSET(cmsvc_socket, &recv_set)) {
			ret = cmsvc_create_session(&session, cmsvc_socket);
			if (ret == eRET_SUCCESS) {
				list_add(&session->sibling, &session_list);
			} else {
				sysmgr_err("cmsvc_create_session() failed, ret: %d.\n", ret);
			}
		}

		list_for_each_safe(each, next, &session_list) {
			session = list_entry(each, tcm_session, sibling);

			clock_gettime(CLOCK_MONOTONIC, &ts);

			if (FD_ISSET(session->conn_fd, &recv_set)) {
				session->timestamp = ts.tv_sec;
				ret = cmsvc_proc_session(session);
				if (ret != eRET_SUCCESS) {
					sysmgr_err("cmsvc_proc_session() failed.\n");
				}
			} else {
				if (ts.tv_sec > session->timestamp + SESSION_IDLE_CHECK_TIME) {
					/* check connect status */
					rc = recv(session->conn_fd, (void *)&rx_buff, sizeof(rx_buff), MSG_DONTWAIT);
					if (rc == 0) {
						sysmgr_err("recv() peer shutdown!\n");
						session->shutdown = eBOOL_TRUE;
					}
				}
			}

			if (session->shutdown) {
				list_del(each);
				cmsvc_destroy_session(session);
			}
		}
	}

out:

	list_for_each_safe(each, next, &session_list) {
		session = list_entry(each, tcm_session, sibling);
		close(session->conn_fd);
		list_del(each);
		free(session);
	}

	cmsvc_svc_socket_clean(cmsvc_socket);

	sysmgr_trace_exit();

#ifdef CMSVC_USE_TASK
	pthread_exit(NULL);
#else /* CMSVC_USE_TASK */
	return (NULL);
#endif /* CMSVC_USE_TASK */

}

e_ret cmsvc_svc_socket_init(int *cmsvc_socket)
{
	int rc;
	int fd;
	int len;
	long fd_flags;
	struct sockaddr_un unix_addr;

	sysmgr_trace_enter();

	*cmsvc_socket = 0;
	fd = socket(AF_UNIX, SOCK_STREAM, 0);
	if (fd < 0) {
		sysmgr_err("socket() failed, err: %d,[%s]\n", errno, strerror(errno));
		return (eRET_FAILURE);
	}

	fd_flags = fcntl(fd, F_GETFD, 0);
	fcntl(fd, F_SETFD, fd_flags | FD_CLOEXEC);

	fd_flags = fcntl(fd, F_GETFL, 0);
	fcntl(fd, F_SETFL, fd_flags | O_NONBLOCK);

	unlink(CONFIG_CM_SOCKET);

	memset(&unix_addr, 0, sizeof(unix_addr));
	unix_addr.sun_family = AF_UNIX;
	strcpy(unix_addr.sun_path, CONFIG_CM_SOCKET);
	len = sizeof(unix_addr);

	rc = bind(fd, (struct sockaddr *)&unix_addr, len);
	if (rc < 0) {
		sysmgr_err("bind() failed, err: %d,[%s]\n", errno, strerror(errno));
		close(fd);
		unlink(CONFIG_CM_SOCKET);
		sysmgr_trace_exit();
		return (eRET_FAILURE);
	}

	rc = chmod(CONFIG_CM_SOCKET, 0666);
	if (rc < 0) {
		sysmgr_err("chmod() failed, err: %d,[%s]\n", errno, strerror(errno));
		close(fd);
		unlink(CONFIG_CM_SOCKET);
		sysmgr_trace_exit();
		return (eRET_FAILURE);
	}

	rc = listen(fd, 32);
	if (rc < 0) {
		sysmgr_err("listen() failed, err: %d,[%s]\n", errno, strerror(errno));
		close(fd);
		unlink(CONFIG_CM_SOCKET);
		sysmgr_trace_exit();
		return (eRET_FAILURE);
	}

	*cmsvc_socket = fd;

	sysmgr_trace_exit();

	return (eRET_SUCCESS);

}

e_ret cmsvc_svc_socket_clean(int cmsvc_socket)
{

	sysmgr_trace_enter();

	if (cmsvc_socket > 0) {
		close(cmsvc_socket);
		unlink(CONFIG_CM_SOCKET);
	}

	sysmgr_trace_exit();

	return (eRET_SUCCESS);

}

e_ret cmsvc_recv_msg(tcm_session * session, struct timeval * timeout)
{

	e_ret ret;
	int rc;
	int fd;
	int msg_len;
	int buf_len;
	int recv_len;
	e_bool got_msg_hdr;
	unsigned char *rx_buff;

	sysmgr_trace_enter();

	ret = eRET_SUCCESS;
	recv_len = 0;
	fd = session->conn_fd;
	session->msg_len = 0;
	msg_len = buf_len = session->buf_len;
	rx_buff = (unsigned char *)session->msg_buf;
	got_msg_hdr = eBOOL_FALSE;

	do {
		rc = recv(fd, (void *)&rx_buff[recv_len], buf_len - recv_len, 0);
		if (rc > 0) {
			recv_len += rc;
			if (!got_msg_hdr) {
				if (recv_len >= sizeof(tcm_msg)) {
					got_msg_hdr = eBOOL_TRUE;
					msg_len = sizeof(tcm_msg) + session->msg_buf->header.opt_body_len;
					if (buf_len < msg_len) {
						buf_len = msg_len;
						rx_buff = malloc(buf_len);
						if (rx_buff == NULL) {
							sysmgr_err("malloc(%d) failed, err: %d,[%s]\n",
								   buf_len, errno, strerror(errno));
							ret = eRET_NO_RESOURCE;
							goto done;
						} else {
							memcpy(rx_buff, session->msg_buf, recv_len);
							free(session->msg_buf);
							session->msg_buf = (void *)rx_buff;
							session->buf_len = buf_len;
						}
					}
				}
			}
			session->msg_len = recv_len;
		} else if (rc < 0) {
			if ((errno == EAGAIN) || (errno == EINTR)
			    || (errno == EWOULDBLOCK)) {
				continue;
			} else {
				sysmgr_err("recv() return %d err: %d,[%s]\n", rc, errno, strerror(errno));
				ret = eRET_FAILURE;
				goto done;
			}
		} else {
			sysmgr_err("recv() peer shutdown!\n");
			session->shutdown = eBOOL_TRUE;
			ret = eRET_FAILURE;
			break;
		}
	} while (recv_len < msg_len);

done:
	if (ret == eRET_SUCCESS) {
		session->transid = session->msg_buf->header.transid++;
	}
	sysmgr_trace_exit();
	return (ret);
}

e_ret cmsvc_send_msg(tcm_session * session, struct timeval * timeout)
{

	e_ret ret;
	int rc;
	int fd;
	int msg_len;
	int send_len;
	int select_counter;
	unsigned char *tx_buff;
	fd_set send_set;

	sysmgr_trace_enter();

	ret = eRET_SUCCESS;
	send_len = 0;
	fd = session->conn_fd;
	tx_buff = (unsigned char *)session->msg_buf;
	msg_len = session->msg_len;

	do {
		FD_ZERO(&send_set);
		FD_SET(fd, &send_set);

		select_counter = select(fd + 1, NULL, &send_set, NULL, timeout);
		if (select_counter == 0) {
			if (timeout != NULL) {
				/* Time out... */
				sysmgr_dbg("select() timeout for send() err: %d,[%s]\n", errno, strerror(errno));
				ret = eRET_TIMEOUT;
				goto done;
			} else {
				continue;
			}
		} else if (select_counter < 0) {
			/* select be interrupt? */
			if (errno != EINTR) {
				sysmgr_dbg("select() return %d, err: %d,[%s]\n",
					   select_counter, errno, strerror(errno));
				ret = eRET_FAILURE;
				goto done;
			} else {
				continue;
			}
		}

		if (FD_ISSET(fd, &send_set)) {
			rc = send(fd, (const void *)&tx_buff[send_len], msg_len - send_len, 0);
			if (rc <= 0) {
				if ((errno == EAGAIN) || (errno == EINTR)
				    || (errno == EWOULDBLOCK)) {
					continue;
				} else {
					sysmgr_err("send() return %d, err: %d,[%s]\n", rc, errno, strerror(errno));
					ret = eRET_FAILURE;
					goto done;
				}
			} else {
				send_len += rc;
			}
		}
	} while (send_len < msg_len);

done:
	if (ret == eRET_SUCCESS) {
		session->transid++;
	}
	sysmgr_trace_exit();
	return (ret);
}

e_ret cmsvc_proc_msg(tcm_session * session)
{

	e_ret ret;
	int i;
	int id;
	tcm_session_body *body;
	tcm_msg *msg = session->msg_buf;

	sysmgr_trace_enter();

	switch (msg->header.opt) {
	case eCM_OPERATION_CREATE_SESSION:
		id = SESSION_ID_TMP;
		for (i = 0; i < SESSION_ID_MAX; i++) {
			if (id_table[i] == SESSION_ID_TMP) {
				id = i + 1;
				id_table[i] = id;
				break;
			}
		}
		if (id != SESSION_ID_TMP) {
			body = (tcm_session_body *) msg->opt_body;
			str_safe_cpy(session->name, body->from, CM_SESSION_NAME_SIZE - 1);
			session->id = id;
			ret = eRET_SUCCESS;
		} else {
			session->shutdown = eBOOL_TRUE;
			ret = eRET_NO_RESOURCE;
		}
		msg->header.opt_body_len = 0;
		break;

	case eCM_OPERATION_DESTORY_SESSION:
		if ((msg->header.sessionid != SESSION_ID_TMP) && (msg->header.sessionid <= SESSION_ID_MAX)) {
			i = msg->header.sessionid - 1;
			id_table[i] = SESSION_ID_TMP;
			session->shutdown = eBOOL_TRUE;
			ret = eRET_SUCCESS;
		} else {
			ret = eRET_NOT_EXIST;
		}
		msg->header.opt_body_len = 0;
		break;

	case eCM_OPERATION_REQUEST:
		ret = cmsvc_proc_request(session);
		break;

	case eCM_OPERATION_NOTIFY:
		msg->header.opt_body_len = 0;
		ret = eRET_INVALID_STATE;
		break;

	default:
		sysmgr_err("Unkonw operation %d\n", msg->header.opt);
		msg->header.opt_body_len = 0;
		msg->header.ret = eRET_INVALID_ARG;
		ret = eRET_INVALID_ARG;
		break;
	}

	msg->header.ret = ret;
	msg->header.dir = eCM_DIR_REPLY;
	msg->header.sessionid = session->id;

	session->msg_len = sizeof(tcm_msg) + msg->header.opt_body_len;

	sysmgr_trace_exit();

	return (ret);

}

e_ret cmsvc_create_session(tcm_session ** session, int svc_socket)
{

	e_ret ret;
	int fd;
	long fd_flags;
	socklen_t addr_len;
	struct sockaddr_un addr;
	tcm_session *new_session;
	struct timespec ts;

	sysmgr_trace_enter();

	*session = new_session = NULL;
	memset(&addr, 0, sizeof(addr));
	memset(&addr_len, 0, sizeof(addr_len));
	fd = accept(svc_socket, (struct sockaddr *)&addr, &addr_len);
	if (fd < 0) {
		sysmgr_err("accept() failed, err: %d,[%s].\n", errno, strerror(errno));
		ret = eRET_FAILURE;
		goto err;
	}

	new_session = malloc(sizeof(tcm_session));
	if (new_session == NULL) {
		ret = eRET_NO_RESOURCE;
		goto err;
	}
	memset(new_session, 0, sizeof(tcm_session));

	new_session->msg_len = 0;
	new_session->buf_len = CM_MSG_BUFF_SIZE;
	new_session->msg_buf = malloc(new_session->buf_len);
	if (new_session->msg_buf == NULL) {
		ret = eRET_NO_RESOURCE;
		goto err;
	}
	memset(new_session->msg_buf, 0, new_session->buf_len);

	fd_flags = fcntl(fd, F_GETFL, 0);
	fcntl(fd, F_SETFL, fd_flags | O_NONBLOCK);

	clock_gettime(CLOCK_MONOTONIC, &ts);

	new_session->id = SESSION_ID_TMP;
	new_session->conn_fd = fd;
	new_session->addr = addr;
	new_session->addr_len = addr_len;
	new_session->timestamp = ts.tv_sec;

	*session = new_session;

	pf_dep_cfg_init();

	return (eRET_SUCCESS);

err:
	if (fd >= 0) {
		close(fd);
	}
	if (new_session != NULL) {
		if (new_session->msg_buf != NULL) {
			free(new_session->msg_buf);
			new_session->msg_buf = NULL;
			new_session->buf_len = 0;
		}
		free(new_session);
		new_session = NULL;
	}

	sysmgr_trace_exit();
	return (ret);

}

void cmsvc_destroy_session(tcm_session * session)
{
	sysmgr_trace_enter();

	if (session != NULL) {
		if (session->id != SESSION_ID_TMP) {
			pf_dep_cfg_save();
			id_table[session->id - 1] = SESSION_ID_TMP;
		}

		if (session->conn_fd > 0) {
			close(session->conn_fd);
		}

		if (session->msg_buf != NULL) {
			free(session->msg_buf);
			session->msg_buf = NULL;
			session->buf_len = 0;
		}
		free(session);
		session = NULL;
	}
	sysmgr_trace_exit();
}

e_ret cmsvc_proc_session(tcm_session * session)
{
	e_ret ret;
	struct timeval timeout;

	sysmgr_trace_enter();

	timeout.tv_sec = RECV_MSG_TIMEOUT_SECOND;
	timeout.tv_usec = RECV_MSG_TIMEOUT_USECOND;
	ret = cmsvc_recv_msg(session, &timeout);
	if (ret != eRET_SUCCESS) {
		sysmgr_info("cmsvc_recv_msg() failed, ret: %d.\n", ret);
		goto done;
	}

	ret = cmsvc_proc_msg(session);
	if (ret != eRET_SUCCESS) {
		sysmgr_info("cmsvc_proc_msg() failed, ret: %d.\n", ret);
	}

	timeout.tv_sec = SEND_MSG_TIMEOUT_SECOND;
	timeout.tv_usec = SEND_MSG_TIMEOUT_USECOND;
	ret = cmsvc_send_msg(session, &timeout);
	if (ret != eRET_SUCCESS) {
		sysmgr_info("cmsvc_send_msg() failed, ret: %d.\n", ret);
		goto done;
	}

done:
	sysmgr_trace_exit();
	return (ret);

}

e_ret cmsvc_proc_request(tcm_session * session)
{

	int i;
	e_ret ret = eRET_SUCCESS;
	tcm_msg *msg;
	struct list_head proc_list;
	struct list_head *each, *next;
	tcm_proc_list_node *node;
	tcm_req_body *req;
	tcm_rsp_body *rsp;
	unsigned int rsp_len;
	void *rsp_buf;

	sysmgr_trace_enter();

	msg = session->msg_buf;

	INIT_LIST_HEAD(&proc_list);

	rsp_len = 0;

	for (i = 0; i < msg->header.opt_counter; i++) {
		req = (tcm_req_body *) msg->opt_body;
		node = malloc(sizeof(tcm_proc_list_node));
		if (node == NULL) {
			msg->header.opt_body_len = 0;
			ret = eRET_NO_RESOURCE;
			goto out;
		}
		node->req = NULL;
		node->rsp = NULL;
		list_add_tail(&node->list, &proc_list);

		node->req = malloc(sizeof(*req) + req->req_body_len);
		if (node->req == NULL) {
			msg->header.opt_body_len = 0;
			ret = eRET_NO_RESOURCE;
			goto out;
		}

		memcpy(node->req, req, sizeof(*req) + req->req_body_len);

	}

	list_for_each(each, &proc_list) {
		node = list_entry(each, tcm_proc_list_node, list);
		req = (tcm_req_body *) node->req;

		switch (req->module) {
		case eCM_MODULE_SYS_ADMIN:
			ret = sys_admin_proc_req(req, &rsp);
			break;

		case eCM_MODULE_DATA_PATH:
			ret = data_path_proc_req(req, &rsp);
			break;

		case eCM_MODULE_EXTRA_APP:
			ret = extra_app_proc_req(req, &rsp);
			break;

		default:
			rsp = NULL;
			ret = eRET_NOT_EXIST;
			break;
		}

		if (rsp == NULL) {
			node->rsp = NULL;
			rsp = malloc(sizeof(tcm_rsp_body));
			if (rsp == NULL) {
				msg->header.opt_body_len = 0;
				ret = eRET_NO_RESOURCE;
				goto out;
			}
			rsp->rsp_body_len = 0;
		}

		rsp->module = req->module;
		rsp->ret = ret;

		node->rsp = rsp;
		rsp_len += sizeof(tcm_rsp_body) + rsp->rsp_body_len;
	}

	if (rsp_len >= session->buf_len) {
		rsp_buf = malloc(sizeof(tcm_msg) + rsp_len);
		if (rsp_buf == NULL) {
			msg->header.opt_body_len = 0;
			ret = eRET_NO_RESOURCE;
			goto out;
		}
		memcpy(rsp_buf, session->msg_buf, sizeof(tcm_msg));
		free(session->msg_buf);
		session->msg_buf = rsp_buf;
		session->buf_len = rsp_len;
	}

	msg->header.opt_body_len = rsp_len;
	rsp = (tcm_rsp_body *) session->msg_buf->opt_body;

	rsp_len = 0;
	list_for_each(each, &proc_list) {
		node = list_entry(each, tcm_proc_list_node, list);
		rsp_len += sizeof(tcm_rsp_body) + node->rsp->rsp_body_len;
		memcpy(rsp, node->rsp, sizeof(tcm_rsp_body) + node->rsp->rsp_body_len);
		rsp = (tcm_rsp_body *) (session->msg_buf->opt_body + rsp_len);
	}
	ret = eRET_SUCCESS;

out:

	list_for_each_safe(each, next, &proc_list) {
		node = list_entry(each, tcm_proc_list_node, list);

		list_del(each);

		if (node->req != NULL) {
			free(node->req);
		}
		if (node->rsp != NULL) {
			free(node->rsp);
		}

		free(node);
	}

	sysmgr_trace_exit();

	return (ret);

}
